package com.smsc.headend.task.engine.executor.impl.deviceconfig;

import cn.hutool.json.JSONUtil;
import com.google.common.base.Preconditions;
import com.smsc.headend.module.asset.dto.KeyTransferDTO;
import com.smsc.headend.module.data.entity.MeterRole;
import com.smsc.headend.module.data.enums.OperationName;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.AtomicTask;
import com.smsc.headend.module.task.dto.Task;
import com.smsc.headend.module.task.dto.TaskResult;
import com.smsc.headend.task.engine.dao.TaskDAO;
import com.smsc.headend.task.engine.exception.UDISTaskExecutorException;
import com.smsc.headend.task.engine.executor.TaskExecutor;
import com.smsc.headend.task.engine.handler.impl.deviceconfig.KeyTransferActionHandler;
import com.smsc.headend.module.task.dto.AtomicTaskState;
import com.smsc.headend.task.engine.service.ConnectorService;
import com.smsc.headend.task.engine.service.FeignAssetManagementService;
import com.smsc.headend.task.engine.service.asset.AssetService;
import com.smsc.headend.task.engine.service.task.TaskPacketService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;

import java.nio.charset.Charset;
import java.util.Map;

/**
 * @Description
 * @Author W26102
 * @Date 2021/8/2 15:08
 * @Sine 1.0
 **/
@Slf4j
public class KeyTransferConfirmExecutor implements TaskExecutor {

    public static final String VALUE = "value";
    public static final String MODEL_ID = "modelId";
    public static final String GLOBAL_KEY_TRANSFER = "global_key_transfer";
    @Autowired
    AssetService assetService;
    @Autowired
    FeignAssetManagementService assetManagementRemoteService;
    @Autowired
    TaskPacketService packetService;
    @Autowired
    ConnectorService connectorService;
    @Autowired
    TaskDAO taskDAO;

    @Autowired
    KafkaTemplate kafkaTemplate;

    @Override
    public AtomicTaskState execute(Task task, AtomicTask atomicTask) throws UDISTaskExecutorException {
        log.info("task no:{},atomic no:{}", task.getTaskNo(), atomicTask.getAtomicTaskNo());
        Map<String, Object> paramMap = getParameterMap(task, atomicTask);
        KeyTransferDTO keyTransferDTO = JSONUtil.toBean((String) paramMap.get(VALUE), KeyTransferDTO.class);
        Long modelId = Long.valueOf((String) paramMap.get(MODEL_ID));
        MeterRole meterRole = assetManagementRemoteService.getRoleByRoleName(modelId, keyTransferDTO.getRoleName());
        Preconditions.checkNotNull(meterRole);

        connectorService.sendForReCreateAA(task, atomicTask, meterRole.getRoleId().intValue(), true);

        Class handlerClass = KeyTransferActionHandler.class;
        return createAtomicState(handlerClass);
    }

    @Override
    public Boolean handleFailure(Task task, AtomicTask atomicTask, String message) {
        log.info("error {},task no:{},atomic no:{} ", message, task.getTaskNo(), atomicTask.getAtomicTaskNo());
        OperationName operationName = OperationName.forId(atomicTask.getOperationId());
        taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), operationName.getName(), message, 600L, false);

        TaskResult taskResult = new TaskResult();
        taskResult.setMessageId(task.getMessageId());
        taskResult.setTaskNo(task.getTaskNo());
//        Map<String, Object> taskResultMap = JSONUtil.toBean(task.getParamJson(), Map.class);
//        Meter meter = assetService.getMeterById(task.getMeterId());
//        taskResultMap.put("meterModelId", meter.getModelId());
        taskResult.setExceptionCause(message);
//        taskResult.setParamJson(JSONUtil.toJsonStr(taskResultMap));
        kafkaTemplate.send(TaskKafkaTopic.TASK_COMPLETE_RESULT_FROM_UTE_TO_ADAPTER, JSONUtil.toJsonStr(taskResult).getBytes(Charset.forName("utf-8")));

        return true;
    }
}
